package com.wt.dwh.samples.flink;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Collection;

public class ProcessFunctionExample {
    public static void main(String[] args) throws  Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<UserClickEvent> source = env.fromCollection(UserClickEventSource.ELS).name("in-memory-input");
//        stream.assignTimestampsAndWatermarks(WatermarkStrategy.<UserClickEvent>forMonotonousTimestamps()
//                .withTimestampAssigner(new SerializableTimestampAssigner<UserClickEvent>() {
//                    @Override
//                    public long extractTimestamp(UserClickEvent event, long recordTimestamp) {
//                        return event.getTimestamp();
//                    }
//                })
//        );
        SingleOutputStreamOperator<UserClickEvent>  watermarkDataSource = source.assignTimestampsAndWatermarks(WatermarkStrategy.<UserClickEvent>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, recordTimestamp) -> event.getTimestamp())
        );

        SingleOutputStreamOperator<String> processDataSource = watermarkDataSource.process(new ProcessFunction<UserClickEvent, String>() {
            @Override
            public void processElement(UserClickEvent value, Context ctx, Collector<String> out) throws Exception {
                ctx.timerService().currentWatermark();
                ctx.output(new OutputTag<UserClickEvent>("outPut") {}, value);
                getRuntimeContext().getIndexOfThisSubtask();
//                out.collect(value.toString());
            }
        });
        processDataSource.print("ProcessFunction");

        SingleOutputStreamOperator<Object> keyedProcessDatasource = watermarkDataSource.keyBy(e -> e.getUrl())
                .process(new KeyedProcessFunction<String, UserClickEvent, Object>() {
                    @Override
                    public void processElement(UserClickEvent value, Context ctx, Collector<Object> out) throws Exception {
                        out.collect(value);
                    }
                });
//        keyedProcessDatasource.print("KeyedProcessFunction");

        SingleOutputStreamOperator<Object> processWindowDataSource = watermarkDataSource.keyBy(e -> e.getUrl())
                .window(TumblingEventTimeWindows.of(Time.milliseconds(2000)))
                .process(new ProcessWindowFunction<UserClickEvent, Object, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<UserClickEvent> elements, Collector<Object> out) throws Exception {
                        System.out.println("ProcessWindowFunction->process->  "+s);
                    }
                });
//        processWindowDataSource.print("ProcessWindowFunction");
        env.execute();
    }
}
